package com.hank.activemq.springboot;

import static org.hamcrest.CoreMatchers.instanceOf;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQBytesMessage;

public class MQConsumer {
	public static void main(String[] args) throws JMSException {
		// 连接到ActiveMQ服务器
//		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://server1.com:61618");
        // 使用容错连接 failover:(tcp://server1.com:61616,tcp://1server1.com:61617,tcp://server1.com:61618)?randomize=false
        //randomize 使用随机连接，达到负载均衡的目的
        String brokerURL="failover:(tcp://server1.com:61616,tcp://server1.com:61617,tcp://server1.com:61618)?randomize=false";
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL);
		Connection connection = factory.createConnection();
		connection.start();
		Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
		// 创建主题
		Queue topicA = session.createQueue("Consumer.A.VirtualTopic.TEST");
		Queue topicB = session.createQueue("Consumer.B.VirtualTopic.TEST");
		// 创建消费者A组1-3号订阅
//		String port = "61618-";
		createConsumerQueue(session.toString()+ "A1", topicA, session);
//		createConsumerQueue(port + "A2", topicA, session);
		// createConsumerQueue("A3", topicA, session);
		// 创建消费者B组1-3号订阅
		// createConsumerQueue("B1", topicB, session);
		// createConsumerQueue("B2", topicB, session);
		// createConsumerQueue("B3", topicB, session);
	}

	private static void createConsumerQueue(String name, Queue topic, Session session) throws JMSException {
		// 创建消费者A组订阅
		MessageConsumer consumerA1 = session.createConsumer(topic);
		consumerA1.setMessageListener(new MessageListener() {
			@Override
			public void onMessage(Message msg) {
				try {
					if (msg instanceof TextMessage) {
						TextMessage tm = (TextMessage) msg;
						System.out.println("Received message " + name + ": " + tm.getText() + ":"
								+ tm.getStringProperty("property"));

					} else if (msg instanceof ActiveMQBytesMessage) {
						ActiveMQBytesMessage bytesMessage = (ActiveMQBytesMessage) msg;
						if (bytesMessage != null) {
							byte[] bt = new byte[(int) bytesMessage.getBodyLength()];
							bytesMessage.readBytes(bt);
							System.out.println("Received message " + name + ": " +new String(bt));
						}
					}
					Thread.sleep(100);
				} catch (JMSException e1) {
					e1.printStackTrace();
				} catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
		});
	}
}
